package com.czk.server.common;

import io.netty.channel.*;

import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

/**
 * @Author:ChenZhangKun
 * @Date: 2021/5/7 10:07
 */
public abstract class MyServerCommonHandler extends SimpleChannelInboundHandler<String> {
    protected boolean sendFlag;
    private Runnable counterTask;
    /**
     * cas计数，防止线程紊乱
     */
    private AtomicLong consumeMsgLength = new AtomicLong();
    private Long priorProgress;

    /**
     * @param channelHandlerContext
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        System.out.println(" NettyServer接收到消息：" + s);
    }

    /**
     * 一个连接来的时候触发
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        counterTask = () -> {
            while (true) {
                try {
                    // 休眠
                    Thread.sleep(500);
                    // 拿到长度
                    long length = consumeMsgLength.getAndSet(0);
                    if (length == 0) continue;
                    System.out.println("数据发送速率（kb/s）" + length);
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        };
        super.handlerAdded(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        sendData(ctx);
        // 启动线程
        new Thread(counterTask).start();
    }

    /**
     * 子类来实现
     *
     * @param ctx
     */
    protected abstract void sendData(ChannelHandlerContext ctx);

    protected ChannelProgressivePromise getChannelProgressivePromise(ChannelHandlerContext ctx, Consumer<ChannelProgressiveFuture> completedAction) {
        ChannelProgressivePromise channelProgressivePromise = ctx.newProgressivePromise();
        // 添加监听器
        channelProgressivePromise.addListener(new ChannelProgressiveFutureListener() {
            @Override
            public void operationProgressed(ChannelProgressiveFuture channelProgressiveFuture, long l, long l1) throws Exception {
                consumeMsgLength.addAndGet(l - priorProgress);
                priorProgress = l;
            }

            @Override
            public void operationComplete(ChannelProgressiveFuture channelProgressiveFuture) throws Exception {
                sendFlag = false;
                if (channelProgressiveFuture.isSuccess()) {
                    System.out.println("消息发送成功！");
                    priorProgress -= 10;
                    Optional.ofNullable(completedAction).ifPresent(action -> action.accept(channelProgressiveFuture));
                } else {
                    System.out.println("发送失败！");
                    channelProgressiveFuture.cause().printStackTrace();
                }
            }
        });
        return channelProgressivePromise;
    }

}
